package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.CartAddUuBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 从 Kafka 读取用户加购明细数据，统计 各窗口当日加购独立用户数 ，写入 Doris。
 */
public class DwsTradeCartAddUuWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeCartAddUuWindow().start(10026, 4, "dws_trade_cart_add_uu_window", Constant.TOPIC_DWD_TRADE_CART_ADD);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 读取数据流 stream
        // 转换数据结构为JSONObject 并 设置水位线
        SingleOutputStreamOperator<JSONObject> mapDs = stream.map(JSON::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
                                .withTimestampAssigner((jsonObj, ts) -> (jsonObj.getLong("ts") * 1000))
                                .withIdleness(Duration.ofSeconds(120L))
                );
        // mapDs.print();

        // 按照用户id分组
        SingleOutputStreamOperator<CartAddUuBean> processDs = mapDs.keyBy(jsonObject -> jsonObject.getString("user_id"))
                .process(
                        new KeyedProcessFunction<String, JSONObject, CartAddUuBean>() {
                            private ValueState<String> lastCartAddState;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                RuntimeContext runtimeContext = getRuntimeContext();
                                lastCartAddState = runtimeContext.getState(
                                        new ValueStateDescriptor<String>("lastCartAddStateDesc", Types.STRING)
                                );
                            }

                            @Override
                            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, CartAddUuBean>.Context context, Collector<CartAddUuBean> collector) throws Exception {
                                // 过滤出独立用户加购记录  使用状态编程 将用户最后一次加购的日期维护到状态中
                                // 如果末次加购日期为null或者不为当日, 数据保留, 维护状态; 否则丢弃

                                // 获取状态
                                String lastCartAddDateTime = lastCartAddState.value();
                                // 获取当日
                                String today = DateFormatUtil.tsToDateTime(jsonObject.getLong("ts") * 1000);
                                if (!today.equals(lastCartAddDateTime)) {
                                    lastCartAddState.update(today);
                                    collector.collect(
                                            CartAddUuBean.builder()
                                                    .stt("")
                                                    .edt("")
                                                    .curDate("")
                                                    .cartAddUuCt(1L)
                                                    .build()
                                    );
                                }
                            }
                        }
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(5L)))
                .reduce(
                        // 开窗聚合  此时每条数据为一个独立用户数
                        new ReduceFunction<CartAddUuBean>() {
                            @Override
                            public CartAddUuBean reduce(CartAddUuBean t1, CartAddUuBean t2) throws Exception {
                                t1.setCartAddUuCt(t1.getCartAddUuCt() + t2.getCartAddUuCt());
                                return t1;
                            }
                        },
                        new ProcessAllWindowFunction<CartAddUuBean, CartAddUuBean, TimeWindow>() {
                            @Override
                            public void process(ProcessAllWindowFunction<CartAddUuBean, CartAddUuBean, TimeWindow>.Context context, Iterable<CartAddUuBean> iterable, Collector<CartAddUuBean> collector) throws Exception {
                                CartAddUuBean bean = iterable.iterator().next();
                                bean.setStt(DateFormatUtil.tsToDateTime(context.window().getStart()));
                                bean.setEdt(DateFormatUtil.tsToDateTime(context.window().getEnd()));
                                bean.setCurDate(DateFormatUtil.tsToDate(context.window().getStart()));
                                collector.collect(bean);
                            }
                        }
                );

        // processDs.print("processDs");

        // 写入到doris中
        processDs.map(
                new BeanToJsonMapFunction<>()
        ).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DORIS_DATABASE, Constant.DWS_TRADE_CART_ADD_UU_WINDOW));








    }
}
